package com.yixun.service;

import com.alibaba.fastjson.JSONObject;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.yixun.component.RedisService;
import com.yixun.constant.SystemConstant;
import com.yixun.model.BreakdownInfo;
import com.yixun.utils.SiteUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

import java.lang.reflect.Type;
import java.util.*;

@Service
@Slf4j
public class ThanNullService {

    private static Gson gson = new Gson();

    @Autowired
    RedisService redisService;

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Async("asyncPool")
    public void thanNull(String json) {
        Type type = new TypeToken<List<List<String>>>() {}.getType();
        //15s内的温度数据集合(大概700毫秒一条)，//30秒接收C#一条数据，700毫秒消费一条
        List<List<String>> siteData = gson.fromJson(json, type);
        for (List<String> tempList : siteData){
            //逻辑代码
            logic(tempList);
        }
    }

    private void logic(List<String> siteData){
        String siteNo = siteData.get(0);
        List<String> siteTemp = siteData.subList(2,siteData.size());

        //按通道（20个一通道）进行分组  key为通道号
        Map<Integer, List<String>> gSiteTemps = SiteUtils.groupByAisle(siteTemp);

        //收集通道中测温点数据全部为空的通道
        List<Integer> aisles = new ArrayList<>();
        gSiteTemps.forEach((aisle,gSiteTemp)->{
            Integer size = gSiteTemp.size();
            Integer nullSize = Math.toIntExact(gSiteTemp.stream().filter(ObjectUtils::isEmpty).count());

            if(size.equals(nullSize)){
                aisles.add(aisle);
            }
        });

        //超过12个通道数据为空，报故障
        BreakdownInfo breakdownInfo = null;

        String breakdownKey = String.format(SystemConstant.BREAKDOWN_THAN_NULL_KEY,siteNo);

        Date currentDate = new Date();

        Boolean exists = redisService.exists(breakdownKey);

        //不报故障的情况 //存在缓存  //把redis缓存的故障删掉 //把此站点关闭故障的信息发给kafka
        if(aisles.size()<12){
            //存在缓存
            if(exists){
                breakdownInfo = JSONObject.parseObject(redisService.get(breakdownKey),BreakdownInfo.class);
                breakdownInfo.setTime(String.valueOf(currentDate.getTime()));
                breakdownInfo.setStatus(BreakdownInfo.STATUS_CLOSED);
                //删掉缓存
                redisService.del(breakdownKey);
                //推送至kafka
                kafkaTemplate.send(SystemConstant.BREAKDOWN_TOPIC,JSONObject.toJSONString(breakdownInfo));
            }

            return;
        }
        //报故障的情况
        //是否存在缓存
        //存在-更新
        //不存在-新增  款村到redis和推送到kafka
        if(exists){
            breakdownInfo = JSONObject.parseObject(redisService.get(breakdownKey),BreakdownInfo.class);
            breakdownInfo.setTime(String.valueOf(currentDate.getTime()));
            breakdownInfo.setStatus(BreakdownInfo.STATUS_WARNING);
            breakdownInfo.setRemark("空数据通道::"+JSONObject.toJSONString(aisles.toArray()));
        }else{
            breakdownInfo = new BreakdownInfo();
            breakdownInfo.setUid(String.valueOf(redisService.incr(SystemConstant.BREAKDOWN_UID)));
            breakdownInfo.setRemark("空数据通道::"+JSONObject.toJSONString(aisles.toArray()));
            breakdownInfo.setSiteNo(siteNo);
            breakdownInfo.setStatus(BreakdownInfo.STATUS_WARNING);
            breakdownInfo.setTime(String.valueOf(currentDate.getTime()));
            breakdownInfo.setType(BreakdownInfo.TYPE_THAN_NULL);
        }
        String breakdownInfoJson = JSONObject.toJSONString(breakdownInfo);
        //更新redis
        redisService.set(breakdownKey,breakdownInfoJson);
        //推送至kafka
        kafkaTemplate.send(SystemConstant.BREAKDOWN_TOPIC,breakdownInfoJson);
    }
}
